package WordConCurrence;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import WordConCurrence.WordConcurrnce.WordConcurrenceMapper;

public class WordConcurrenceReducer extends Reducer<WordPair, IntWritable, WordPair, IntWritable>{
	@Override
	public void reduce(WordPair wordPair, Iterable<IntWritable> frequence,
	Context context) throws IOException, InterruptedException {
		int sum = 0;
		for (IntWritable val : frequence) {
			sum += val.get();
		}
		context.write(wordPair, new IntWritable(sum));
	}
	
	public static void main(String[] args) throws IOException,
	InterruptedException, ClassNotFoundException {
		
		Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
                
        // 检查命令语法
        if (otherArgs.length != 2) {
                System.err.println("Usage: wordcount <in> <out>");
                System.exit(2);
        }
                
        // 定义作业对象
        Job wordConcurrenceJob = new Job(conf, "wordConcurrenceJob");
		
		// 注册分布式类
		wordConcurrenceJob.setJarByClass(WordConcurrnce.class);
		
		//wordConcurrenceJob.getConfiguration().setInt("window",Integer.parseInt(otherArgs[2]));
		
		// 注册Mapper类，注册输出格式类
		wordConcurrenceJob.setMapperClass(WordConcurrenceMapper.class);
		wordConcurrenceJob.setMapOutputKeyClass(WordPair.class);
		wordConcurrenceJob.setMapOutputValueClass(IntWritable.class);
	
		// 注册Reducer类，注册输出格式类
		wordConcurrenceJob.setReducerClass(WordConcurrenceReducer.class);
		wordConcurrenceJob.setOutputKeyClass(WordPair.class);
		wordConcurrenceJob.setOutputValueClass(IntWritable.class);
	
		wordConcurrenceJob.setInputFormatClass(WholeFileInputFormat.class);
		wordConcurrenceJob.setOutputFormatClass(TextOutputFormat.class);
		
		// 设置输入输出路径
		FileInputFormat.addInputPath(wordConcurrenceJob, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(wordConcurrenceJob, new Path(otherArgs[1]));
	
	
		wordConcurrenceJob.waitForCompletion(true);
		System.out.println("finished!");
	}
}
